package com.mashibing.rocketmq02;

import java.util.ArrayList;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;


/**
 * @Author: JLL
 * @Date: 2020/3/3 15:02
 *
 * 这里讲RocketMQ的事务消息，这里测一下 UNKNOW
 *
 *  以下是程序执行结果
 *  =====executeLocalTransaction
 * msg:测试！这是事务消息
 * msg:2408805FE228B31B3DD53B7A4913040F000000B4AAC20DBB597E0000
 * sendResult:SendResult [sendStatus=SEND_OK, msgId=2408805FE228B31B3DD53B7A4913040F000000B4AAC20DBB597E0000, offsetMsgId=null, messageQueue=MessageQueue [topic=xxoo001, brokerName=broker-a, queueId=3], queueOffset=0]
 * 已经停机
 * =====checkLocalTransaction
 * msg:测试！这是事务消息
 * msg:2408805FE228B31B3DD53B7A4913040F000000B4AAC20DBB597E0000
 *
 */
public class Producer {
    public static void main(String[] args)throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("xoxogp1");
        producer.setNamesrvAddr("192.168.150.133:9876");

//        producer.setExecutorService();  发事务消息的时候还可以设置线程池
        // 回调
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
                // 执行 本地事务
                System.out.println("=====executeLocalTransaction");
                System.out.println("msg:" + new String(msg.getBody()));
                //msg是事务消息，所以有
                System.out.println("msg:" + msg.getTransactionId());

                /**
                 * 开始执行事务，一系列的方法。balabala..  注意，  这里面的代码是同步执行的。
                 * 这里拿一个需求举例，   （a b c三个方法肯定想要是要么都成功 要么都失败，比如b写库都失败了，那我还发个毛的短信啊）
                 * -----------这是A程序
                 * a 提交注册信息()  ;
                  *b 写入数据库();
                   c 新用户(这里会联动调用B的service，是通过消息中间件调的) -> 发消息;

                 * -----------这是B程序
                 * 它从中间件里读消息，
                 * 读取消息
                 * 拿到新用户的信息，然后发短信
                 *
                 * 写个伪代码如下
                 */
                try {

                    // 业务
                } catch (Exception e) {
                    // 业务抛异常了， 这里rollback 是把 半消息撤掉。。  正常执行完了，就直接COMMIT_MESSAGE了。
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //然后commit，这里COMMIT_MESSAGE了下面的checkLocalTransaction方法就不会被执行了。
                /**
                 * //这里UNKNOW的话，表示还不确定事务啥时候提交。然后把UNKNOW发给broker。
                 * 然后就会进入到checkLocalTransaction方法，这这个方法里去检查事务是不是成功了。
                 */
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // Broker 端 回调 ，检查事务
                /**
                 * 这个方法是broker发起的回调请求。 而且顺便还会把我们发的业务Message（测试！这是事务消息）给发过来了。
                 * 然后再跑本地事务， 跑成功了就发一个COMMIT_MESSAGE，不成功就发ROLLBACK_MESSAGE （这时候那个 半消息在broker里就被清除了）
                 * 如果发UNKNOW的话，它就等会儿再来。
                 */
                System.out.println("=====checkLocalTransaction");
                System.out.println("msg:" + new String(msg.getBody()));
                System.out.println("msg:" + msg.getTransactionId());

                // 事务执行成功
//                return LocalTransactionState.COMMIT_MESSAGE;
                // 等会儿
                return LocalTransactionState.UNKNOW;
                // 回滚消息
//                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
        producer.start();
        //注意啊，到这一步，  我们的业务message还没发呢。         半消息确认了，才可以让broker里发业务消息

        //这里才是业务消息
        TransactionSendResult sendResult = producer.sendMessageInTransaction(new Message("xxoo001", "测试！这是事务消息".getBytes()), null);
        System.out.println("sendResult:" + sendResult);
        //注意这里不能shutdown，  这根之前的还不一样，之前的关不关都行。这里不行，关了的话，异步回调函数就看不到效果 了。
        //	producer.shutdown();
        System.out.println("已经停机");
    }
}
